package com.chinaums.sftp.tablesource;

import com.chinaums.sftp.internal.connection.SftpHandler;
import com.chinaums.sftp.internal.options.SftpReadOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static com.chinaums.sftp.internal.constants.ConstantValue.SUFFIX_TAR_GZ;
import static com.chinaums.sftp.internal.constants.ConstantValue.SUFFIX_GZ;
import static com.chinaums.sftp.internal.constants.ConstantValue.SUFFIX_ZIP;

/**
 * sftp buffer
 */
public class SftpSeqBufferedReader implements Serializable {
    private static final long serialVersionUID = 1L;


    private static final Logger LOG = LoggerFactory.getLogger(SftpSeqBufferedReader.class);

    private final SftpHandler ftpHandler;

    private final Iterator<String> inter;

    private BufferedReader br;

    private String fileEncoding;

    private String fileName;

    private long fromLine = 0;

    private Map<String, Long> fileReadIndex = new HashMap<>();

    private long fileIndex;

    private final Pattern pattern = Pattern.compile("[a-zA-Z_]*");

    //ftp配置信息
    private final SftpReadOptions readOptions;

    public static boolean isTail = false;


    public String getFileName() {
        return fileName;
    }

    public SftpSeqBufferedReader(SftpHandler ftpHandler, Iterator<String> inter, SftpReadOptions readOptions) {
        this.ftpHandler = ftpHandler;
        this.inter = inter;
        this.readOptions = readOptions;
    }


    /**
     * 解压缩中文件读取文件 第一行与最后一行 异常修复
     *
     * @param line 一行数据
     * @return line  一行数据
     */
    public String ProcessDataFromHeadAndTail(String line) {
        if (line == null) {
            return null;
        }

        // 第一行与最后一行问题解决
        String nameAndSuffix = new File(fileName).getName();
        if (nameAndSuffix.endsWith("tar.gz")) {

            // 获取二进制文件,转化成集合
            ArrayList<Byte> data = new ArrayList<>();

            for (Byte aByte : line.getBytes(StandardCharsets.UTF_8)) {
                data.add(aByte);
            }

            // 获取首行信息
            String headerInfo = "";
            if (data.contains((byte) 0)) {
                List<Byte> headerInfoList = data.subList(0, data.indexOf((byte) 0));
                byte[] headerInfoByte = new byte[headerInfoList.size()];
                for (int i = 0; i < headerInfoList.size(); i++) {
                    headerInfoByte[i] = headerInfoList.get(i);
                }
                headerInfo = new String(headerInfoByte);
            }


            // 获取 文件名信息
            Matcher m = pattern.matcher(nameAndSuffix);
            String tarGzName = m.find() ? m.group(0) : nameAndSuffix.split("\\.")[0].split("_")[0];  // DM_T_SYS_USER_

            byte[] goodDataByte;

            // 它是首行
            if (headerInfo.contains(tarGzName)) {
                // 获取数据
                List<Byte> goodDataList = data.subList(data.lastIndexOf((byte) 0) + 1, data.size());
                goodDataByte = new byte[goodDataList.size()];
                for (int i = 0; i < goodDataList.size(); i++) {
                    goodDataByte[i] = goodDataList.get(i);
                }
                line = new String(goodDataByte, StandardCharsets.UTF_8);

                // 最后一行
            } else if (data.size() >= 1 && data.get(data.size() - 1) == (byte) 0) {
                isTail = true;
                line = null;

                // 去头去尾 后
            } else {
                goodDataByte = new byte[data.size()];
                for (int i = 0; i < data.size(); i++) {
                    goodDataByte[i] = data.get(i);
                }
                line = new String(goodDataByte, StandardCharsets.UTF_8);
            }
        }else if (nameAndSuffix.endsWith(".zip")){
            return line;
        }
        return line;
    }


    /**
     * 读取一行数据
     * @param flag flag
     * @return 一行数据
     */
    public String readLine(boolean flag) throws IOException {

        // 实例流
        if (br == null) {
            // 将已读文件删除
            if (readOptions.isDeleteRead()) {
                try {
                    ftpHandler.deleteAllFilesInDir(fileName, new ArrayList<>());
                } catch (Exception e) {
                    LOG.warn("delete failed file name - " + fileName);
                } finally {
                    fileName = null;
                }
            }
            nextStream();
        }

        if (br != null) {
            if (flag) {
                br.reset();
                String line = br.readLine();
                LOG.warn("reset from line - " + line);
            }
            //TODO 压缩格式处理
            String line = ProcessDataFromHeadAndTail(br.readLine());

            if (line != null) {
                fileIndex += line.getBytes(StandardCharsets.UTF_8).length + readOptions.getByteDelimiterLength();
                fileReadIndex.put(fileName, fileIndex);
            }

            if (line != null) {
                br.mark(line.getBytes(StandardCharsets.UTF_8).length * 2);
            } else {
                close();
                return readLine(false);
            }
            return line;
        } else {
            return null;
        }

    }

    /**
     * 初始化工作流 ...
     */
    private void nextStream() throws IOException {
        if (inter.hasNext()) {
            String file = inter.next();
            fileName = file;
            fileIndex = 0;

            // 实例流
            if (file.endsWith(SUFFIX_TAR_GZ) || file.endsWith(SUFFIX_GZ)) {
                br = ftpHandler.getGZInputStream(fileName, readOptions.getEncoding());
            } else if (file.endsWith(SUFFIX_ZIP)) {
                br = ftpHandler.getZipInputStream(fileName,readOptions.getEncoding());
            } else {
                br = new BufferedReader(new InputStreamReader(ftpHandler.getInputStream(fileName), fileEncoding));
            }

            if (fileReadIndex.containsKey(fileName)) {
                LOG.info(String.format("file - %s skip - %s byte", fileName, br.skip((fileIndex = fileReadIndex.get(fileName)))));
            } else {
                for (int i = 0; i < fromLine; i++) {
                    String skipLine = br.readLine();
                    LOG.info("Skip line:{}", skipLine);
                }
            }
        }
        else {
            br = null;
        }
    }

    public void close() throws IOException {
        if (br != null) {
            br.close();
            br = null;

        }
    }

    /**
     * 设置文件编码...
     */
    public void setFileEncoding(String fileEncoding) {
        this.fileEncoding = fileEncoding;
    }

    /**
     * 是否读首行
     */
    public void setFromLine(long fromLine) {
        this.fromLine = fromLine;
    }


    public Map<String, Long> getFileReadIndex() {
        return fileReadIndex;
    }

    public void setFileReadIndex(Map<String, Long> fileReadIndex) {
        this.fileReadIndex = fileReadIndex;
    }


}
